package com.sl.cdc.modular.debezium;

import cn.hutool.core.codec.Base64;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.druid.util.JdbcUtils;
import com.google.common.collect.Maps;
import com.sl.cdc.api.domain.*;
import com.sl.cdc.api.enums.CdcDataSourceEnum;
import com.zaxxer.hikari.HikariDataSource;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.anyline.data.jdbc.runtime.JDBCRuntimeHolder;
import org.anyline.data.jdbc.util.DataSourceUtil;
import org.anyline.data.runtime.RuntimeHolder;
import org.anyline.entity.DataRow;
import org.anyline.metadata.Schema;
import org.anyline.metadata.Table;
import org.anyline.proxy.ServiceProxy;
import org.anyline.service.AnylineService;

import javax.sql.DataSource;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.function.Function;

@Slf4j

public class CdcDebeziumConsumer implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {


    private TaskInfo taskInfo;

    private AnylineService sourceAnylineService;
    private AnylineService targetAnyLineService;

    public static final String TABLE = "table";


    public CdcDebeziumConsumer(TaskInfo taskInfo) {
        this.taskInfo = taskInfo;
        TaskDataSourceInfo targetInfo = taskInfo.getTargetInfo();
        buildTargetAnyLineService(targetInfo);

        TaskDataSourceInfo sourceInfo = taskInfo.getSourceInfo();
        buildSourceAnyLineService(sourceInfo);


    }

    @SneakyThrows
    private void buildSourceAnyLineService(TaskDataSourceInfo targetInfo) {
        String jdbcUrl = initJdbcUrl(targetInfo);
        if (StrUtil.isBlank(jdbcUrl)) {
            return;
        }
        String user = targetInfo.getUser();
        String password = targetInfo.getPassword();

        String driverClassName = JdbcUtils.getDriverClassName(jdbcUrl);
        sourceAnylineService = buildAnyService(jdbcUrl, driverClassName, user, password);
    }

    @SneakyThrows
    private void buildTargetAnyLineService(TaskDataSourceInfo targetInfo) {
        if (ObjectUtil.isNull(targetInfo)) {
            return;
        }
        String jdbcUrl = initJdbcUrl(targetInfo);
        if (StrUtil.isBlank(jdbcUrl)) {
            return;
        }
        String user = targetInfo.getUser();
        String password = targetInfo.getPassword();

        String driverClassName = JdbcUtils.getDriverClassName(jdbcUrl);
        targetAnyLineService = buildAnyService(jdbcUrl, driverClassName, user, password);
    }


    @Override
    public void handleBatch(List<ChangeEvent<String, String>> recordChangeEvents, DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> recordCommitter) throws InterruptedException {

        ListIterator<ChangeEvent<String, String>> iterator = recordChangeEvents.listIterator();

        while (iterator.hasNext()) {
            ChangeEvent<String, String> record = iterator.next();
            log.debug("Debezium 解析到一条数据 处理前 Key='{}' value='{}'", record.key(), record.value());
//                recordCommitter.markProcessed(record);
            String value = record.value();
            JSONObject entries = JSONUtil.parseObj(value);

            JSONObject payload = entries.getJSONObject("payload");
            JSONObject schema = entries.getJSONObject("schema");

            String op = payload.getStr("op");

            JSONObject source = payload.getJSONObject("source");
            JSONObject after = payload.getJSONObject("after");
            JSONObject before = payload.getJSONObject("before");

            String db = source.getStr("db");//库名
            String table = source.getStr("table");//表名

            if (ObjectUtil.isNotNull(after)) {
                System.out.println("222");
            }

            //op 字段的值 c，u，d，r 分别对应 create，update，delete，read
            if (ObjectUtil.isNull(op)) {
                continue;
            }

            if (StrUtil.equals("op", "r")) {
                System.out.println("rrrr");
            }
            //获取after结构体里面的表数据，封装成json输出
            ConsumerEntityDTO json1 = new ConsumerEntityDTO();

            //整理成大json串输出
            json1.setDb(db);
            json1.setTable(table);
            json1.setType(op);
            json1.setBeforeData(before);
            json1.setAfterData(after);

            try {
                process(json1);
            } catch (Exception e) {
                log.error("报错了", e);
            }
            recordCommitter.markProcessed(record);
        }
        try {
            recordCommitter.markBatchFinished();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @SneakyThrows
    private void process(ConsumerEntityDTO jsonObject) {

        TaskDataSourceInfo targetInfo = taskInfo.getTargetInfo();
        if (ObjectUtil.isNull(targetInfo)) {
            return;
        }
        Function<ConsumerEntityDTO, Map<String, Object>> processFunc = targetInfo.getProcessFunc();
        List<TaskDataBaseInfo> dataBaseList = targetInfo.getDataBaseList();
//        targetInfo.get

        //如果要同步的数据源是走流程自定义直接返回
        if (CollUtil.isEmpty(dataBaseList) || ObjectUtil.isNull(targetAnyLineService)) {
            processFunc.apply(jsonObject);
            return;
        }

        for (TaskDataBaseInfo taskDataBaseInfo : dataBaseList) {

            String code = taskDataBaseInfo.getCode();

            List<TaskTableInfo> tableList = taskDataBaseInfo.getTableList();

            tableList = buildDefaultTableInfo(tableList, jsonObject);

            //开始处理数据了
            dmlTable(taskDataBaseInfo, tableList, jsonObject, processFunc);

        }
    }

    @SneakyThrows
    private void dmlTable(TaskDataBaseInfo taskDataBaseInfo, List<TaskTableInfo> tableList, ConsumerEntityDTO jsonObject, Function<ConsumerEntityDTO, Map<String, Object>> processFunc) {
        String database = taskDataBaseInfo.getCode();
        Schema schema = new Schema();
        schema.setName(database);
        for (TaskTableInfo taskTableInfo : tableList) {
            String code = taskTableInfo.getCode();

            Table table = new Table();
            table.setSchema(schema);
            table.setName(code);
            boolean exists = targetAnyLineService.exists(table);
            if (BooleanUtil.isFalse(exists)) {
                table = sourceAnylineService.metadata().table(code);
                table.setSchema(schema);
                boolean save = targetAnyLineService.save(table);
            }

            table = targetAnyLineService.metadata().table(schema, code, 1);


            //这里开始新增数据
            if (ObjectUtil.isNotNull(processFunc)) {
                Map<String, Object> data = processFunc.apply(jsonObject);
                jsonObject.setAfterData(data);
            }
            Map<String, Object> data = jsonObject.getAfterData();
            if (MapUtil.isEmpty(data)) {
                continue;
            }
            DataRow dataRows = new DataRow();
            dataRows.putAll(data);
            dataRows.setOverride(true);

            String type = jsonObject.getType();
            switch (type) {
                case "c", "r", "u": {
                    targetAnyLineService.save(table, dataRows);
                    break;
                }
                case "d": {
                    targetAnyLineService.delete(table, dataRows);
                    break;
                }
            }
        }
    }

    private List<TaskTableInfo> buildDefaultTableInfo(List<TaskTableInfo> tableList, ConsumerEntityDTO jsonObject) {
        String table = jsonObject.getTable();
        if (CollUtil.isEmpty(tableList)) {
            TaskTableInfo taskTableInfo = new TaskTableInfo();
            taskTableInfo.setCode(table);
            taskTableInfo.setName(table);
            tableList.add(taskTableInfo);
        }
        return tableList;
    }

    private String initJdbcUrl(TaskDataSourceInfo taskInfo) {
        String hostname = taskInfo.getHostname();
        String port = taskInfo.getPort();
        CdcDataSourceEnum sourceDs = taskInfo.getSourceDs();
        if (ObjectUtil.isNull(sourceDs)) {
            return null;
        }
        //jdbc:mysql://101.35.209.150:3306/square-lake?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true&useInformationSchema=true
        return "jdbc:" + sourceDs.getCode() + "://" + hostname + ":" + port;
    }

    private AnylineService buildAnyService(String jdbcUrl, String jdbcDriver, String jdbcUserName, String jdbcPassword) {
        DataSource dataSource = getDataSource(jdbcUrl, jdbcDriver, jdbcUserName, jdbcPassword);

        String encode = Base64.encode(jdbcUrl);
        if (!RuntimeHolder.contains(encode)) {
            JDBCRuntimeHolder.instance().reg(encode, dataSource);
        }
        AnylineService anylineService = ServiceProxy.service(encode);
        return anylineService;
    }


    protected DataSource getDataSource(String jdbcUrl, String jdbcDriver, String jdbcUserName, String jdbcPassword) {
        Map<String, Object> maps = Maps.newConcurrentMap();
        maps.put("url", jdbcUrl);
        maps.put("driver", jdbcDriver);
        maps.put("username", jdbcUserName);
        maps.put("password", jdbcPassword);
        maps.put("type", HikariDataSource.class.getName());
        DataSource dataSource = DataSourceUtil.build(maps);
        return dataSource;
    }

}
